package com.stone.bigdata.flink;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.scala.DataStream;
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * @author fujianjian
 * @date 2023/8/10 10:31
 **/
public class MysqlSourceFunctionV1 {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        TypeInformation<Map> typeInfo = BasicTypeInfo.of(Map.class);
        DataStream<Map> dataInput = env.addSource(new MysqlSource(), typeInfo);
        dataInput.print();

        env.execute();

    }

    public static class MysqlSource extends RichSourceFunction<Map> {
        private volatile boolean IS_RUNNING = true;
        private Connection connection;
        private PreparedStatement ps;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            Class.forName("com.mysql.cj.jdbc.Driver");
            connection = DriverManager.getConnection("jdbc:mysql://10.89.2.119:3306/ihr_core?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8",
                    "ihr_read_gly", "Ihr@zb$2021");
            ps = connection.prepareStatement("select ee.employee_id employeeId, ee.employee_code employeeCode, ee.employee_name employeeName, " +
                    "ee.id_number idNumber, ee.mobile mobile from t_emp_employee ee left join t_org_unit ou on ee.unit_id = ou.unit_id " +
                    "where ou.`path`  like '80000001_80030015_82010921_80037278_%'");
        }

        @Override
        public void run(SourceContext<Map> sourceContext) throws Exception {
            while (IS_RUNNING) {
                ResultSet rs = ps.executeQuery();
                while (rs.next()) {
                    Map<String, String> retMap = Maps.newHashMap();
                    retMap.put("empId", rs.getString(1));
                    retMap.put("empCode", rs.getString(2));
                    retMap.put("empName", rs.getString(3));
                    retMap.put("empIdnum", rs.getString(4));
                    retMap.put("empMobile", rs.getString(5));

                    sourceContext.collect(retMap);
                }
                TimeUnit.SECONDS.sleep(5);
            }
        }

        @Override
        public void cancel() {
            IS_RUNNING = false;
        }

        @Override
        public void close() throws Exception {
            super.close();
            if (Objects.nonNull(ps)) {
                ps.close();
            }
            if (Objects.nonNull(connection)) {
                connection.close();
            }
        }
    }
}
